package com.burt.zookeeper.zkclient.masterchoose;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;

/**
 * Description: master选举
 * User: Burt
 * Date: 2017-08-14
 * Time: 10:42
 */
public class MasterSelector {

    private ZkClient zkClient;

    /**
     * 需要争抢的节点
     */
    private final static String MASTER_PATH = "/master";

    /**
     * 注册节点内容变化
     */
    private IZkDataListener listener;

    /**
     * master节点
     */
    private UserCenter master;

    /**
     * 其它服务器
     */
    private UserCenter server;

    private boolean isRunning = false;

    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

    public MasterSelector(ZkClient zkClient, UserCenter server) {
        System.out.println("["+server+"] 去争抢master权限");
        this.zkClient = zkClient;
        this.server = server;
        this.listener = new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {

            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                //如果结点被删除，发起选主操作
                chooseMaster();
            }
        };
    }

    /**
     * 开始选举
     */
    public void start(){
        if (!isRunning){
            isRunning = true;
            zkClient.subscribeDataChanges(MASTER_PATH,listener);
            chooseMaster();
        }
    }

    /**
     * 停止选举
     */
    public void stop(){
        if (isRunning){
            isRunning = false;
            scheduledExecutorService.shutdown();
            zkClient.unsubscribeDataChanges(MASTER_PATH,listener);
            releaseMaster();
        }
    }

    /**
     * master选举
     */
    public void chooseMaster() {

        if (!isRunning) {
            System.out.println("当前服务没有启动");
            return;
        }
        try {
            zkClient.createEphemeral(MASTER_PATH, server);
            //把server节点赋值给master
            master = server;
            System.out.println(master + "->我现在已经是master，听我号令");

            //定时器
            //master释放(模拟master出现故障)
            scheduledExecutorService.schedule(()->{
                //释放锁
                releaseMaster();

            },2, TimeUnit.SECONDS);
        } catch (ZkNodeExistsException e) {
            //master已存在
            UserCenter userCenter = zkClient.readData(MASTER_PATH, true);
            if (null == userCenter) {
                System.out.println("启动操作: ");
                chooseMaster();
            } else {
                master = userCenter;
            }
        }
    }

    /**
     * 释放锁
     */
    public void releaseMaster(){
        //判断当前是不是master,只有master才需要释放
        if (checkIsMaster()) {
            zkClient.delete(MASTER_PATH);
        }
    }

    /**
     * 判断当前的server是不是master
     * @return
     */
    private boolean checkIsMaster() {
        UserCenter userCenter = zkClient.readData(MASTER_PATH);
        if (userCenter.getMcName().equals(master.getMcName())){
            master = userCenter;
            return true;
        }
        return false;
    }

}
